package org.lhh.server.core.executor;

import org.apache.commons.lang3.StringUtils;
import org.lhh.server.config.web.exception.EtlException;
import org.lhh.server.core.codec.steps.StepStatusCodec;
import org.lhh.server.core.repository.EtlRepository;
import org.pentaho.di.cluster.SlaveServer;
import org.pentaho.di.core.Const;
import org.pentaho.di.core.Result;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleStepException;
import org.pentaho.di.core.logging.*;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.repository.kdr.KettleDatabaseRepository;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransAdapter;
import org.pentaho.di.trans.TransExecutionConfiguration;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.cluster.TransSplitter;
import org.pentaho.di.trans.performance.StepPerformanceSnapShot;
import org.pentaho.di.trans.step.*;
import org.pentaho.di.www.SlaveServerTransStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class TransExecutor implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(TransExecutor.class);

    private static Map<String, TransExecutor> executors = new ConcurrentHashMap<>();
    private Map<StepMeta, String> StepStatusCodecMap = new HashMap<>();
    private Map<StepMeta, RowMetaInterface> previewMetaMap = new HashMap<>();
    private Map<StepMeta, List<Object[]>> previewDataMap = new HashMap<>();
    private Map<StepMeta, StringBuffer> previewLogMap = new HashMap<>();
    private Map<String, List<StepPerformanceSnapShot>> stepPerformanceSnapShotMap = new HashMap<>();
    private String executionId;
    private TransExecutionConfiguration executionConfiguration;
    private TransMeta transMeta;
    private Trans trans;
    private TransSplitter transSplitter;
    private String carteObjectId;
    private boolean finished;
    private long errCount;

    private TransExecutor(TransExecutionConfiguration transExecutionConfiguration, TransMeta transMeta) {
        this.executionId = transMeta.getName();
        this.executionConfiguration = transExecutionConfiguration;
        this.transMeta = transMeta;
    }


    public static TransExecutor initExecutor(TransExecutionConfiguration transExecutionConfiguration, TransMeta transMeta) {
        TransExecutor transExecutor = new TransExecutor(transExecutionConfiguration, transMeta);
        executors.put(transExecutor.getExecutionId(), transExecutor);
        return transExecutor;
    }

    public static  TransExecutor getTranExecutor(String executionId) {
        return executors.get(executionId);
    }


    public static void removeTran(String executionId) {
        executors.remove(executionId);
    }

    @Override
    public void run() {
        KettleDatabaseRepository repository = EtlRepository.getRepository();
        try {
            if (executionConfiguration.isExecutingLocally()) {
                //注入参数
                transMeta.injectVariables(executionConfiguration.getVariables());
                //注入命名参数
                Map<String, String> paramMap = executionConfiguration.getParams();
                Set<String> keys = paramMap.keySet();
                for (String key : keys) {
                    transMeta.setParameterValue(key, Const.NVL(paramMap.get(key), ""));
                }
                transMeta.activateParameters();
                //设置位置
                Map<String, String> arguments = executionConfiguration.getArguments();
                String[] argumentNames = arguments.keySet().toArray(new String[0]);
                Arrays.sort(argumentNames);
                String[] args = new String[argumentNames.length];
                for (int i = 0; i < args.length; i++) {
                    String argumentName = argumentNames[i];
                    args[i] = arguments.get(argumentName);
                }
                trans = new Trans(transMeta);
                trans.setSafeModeEnabled(executionConfiguration.isSafeModeEnabled());
                trans.setGatheringMetrics(executionConfiguration.isGatheringMetrics());
                trans.setLogLevel(executionConfiguration.getLogLevel());
                trans.setReplayDate(executionConfiguration.getReplayDate());
                trans.setRepository(executionConfiguration.getRepository());
                boolean initialized = false;
                try {
                    trans.prepareExecution(args);
                    capturePreviewData(trans, transMeta.getSteps());
                    initialized = true;
                } catch (KettleException e) {
                    logger.error(transMeta.getName() + ": preparing transformation execution failed", e);
                    checkErrorVisuals();
                }

                if (trans.isReadyToStart() && initialized) {
                    trans.addTransListener(new TransAdapter() {
                        public void transFinished(Trans trans) {
                            checkErrorVisuals();
                        }
                    });
                    trans.startThreads();
                    while (!trans.isFinished()) {
                        Thread.sleep(500);
                    }
                    errCount = trans.getErrors();
                } else {
                    checkErrorVisuals();
                    errCount = trans.getErrors();
                }
            } else if (executionConfiguration.isExecutingRemotely()) {
                carteObjectId = Trans.sendToSlaveServer(transMeta, executionConfiguration, repository, repository.getMetaStore());
                SlaveServer remoteSlaveServer = executionConfiguration.getRemoteServer();
                while (true) {
                    SlaveServerTransStatus transStatus = remoteSlaveServer.getTransStatus(transMeta.getName(), carteObjectId, 0);
                    boolean running = transStatus.isRunning();
                    if (!running) {
                        errCount = transStatus.getResult().getNrErrors();
                        break;
                    }
                    Thread.sleep(500);
                }
            } else if (executionConfiguration.isExecutingClustered()) {
                transSplitter = new TransSplitter(transMeta);
                transSplitter.splitOriginalTransformation();
                TransMeta ot = transSplitter.getOriginalTransformation();
                for (String param : ot.listParameters()) {
                    String value = Const.NVL(ot.getParameterValue(param), Const.NVL(ot.getParameterDefault(param), ot.getVariable(param)));
                    if (!StringUtils.isEmpty(value)) {
                        executionConfiguration.getVariables().put(param, value);
                    }
                }
                try {
                    Trans.executeClustered(transSplitter, executionConfiguration);
                } catch (Exception e) {
                    try {
                        Trans.cleanupCluster(repository.getLog(), transSplitter);
                    } catch (Exception ignored) {
                    }
                    throw e;
                }
                Trans.monitorClusteredTransformation(repository.getLog(), transSplitter, null);
                Result result = Trans.getClusteredTransformationResult(repository.getLog(), transSplitter, null);
                errCount = result.getNrErrors();
            }
        } catch (Exception e) {
            logger.error("transformation execution failed！", e);
        } finally {
            finished = true;
        }
    }


    private void capturePreviewData(final Trans trans, List<StepMeta> stepMetas) {
        final StringBuffer loggingText = new StringBuffer();

        // First clean out previous preview data. Otherwise this method leaks memory like crazy.
        //
        previewLogMap.clear();
        previewMetaMap.clear();
        previewDataMap.clear();

        if (executionConfiguration.isExecutingLocally()) {
            final TransMeta transMeta = trans.getTransMeta();

            for (final StepMeta stepMeta : stepMetas) {
                try {
                    final RowMetaInterface rowMeta = transMeta.getStepFields(stepMeta).clone();
                    previewMetaMap.put(stepMeta, rowMeta);
                    final List<Object[]> rowsData = new LinkedList<>();
                    previewDataMap.put(stepMeta, rowsData);
                    previewLogMap.put(stepMeta, loggingText);
                    StepInterface step = trans.findRunThread(stepMeta.getName());
                    if (step != null) {
                        step.addRowListener(new RowAdapter() {
                            @Override
                            public void rowWrittenEvent(RowMetaInterface rowMeta, Object[] row) throws KettleStepException {
                                if (rowsData.size() < 50) {
                                    try {
                                        rowsData.add(rowMeta.cloneRow(row));
                                    } catch (Exception e) {
                                        // Ignore errors
                                    }
                                }
                            }
                        });
                    }
                } catch (Exception e) {
                    loggingText.append(Const.getStackTracker(e));
                }
            }

            // In case there were errors during preview...
            //
            trans.addTransListener(new TransAdapter() {
                @Override
                public void transFinished(Trans trans) throws KettleException {
                    // Copy over the data from the previewDelegate...
                    //
                    if (trans.getErrors() != 0) {
                        // capture logging and store it...
                        //
                        for (StepMetaDataCombi combi : trans.getSteps()) {
                            if (combi.copy == 0) {
                                StringBuffer logBuffer =
                                        KettleLogStore.getAppender().getBuffer(combi.step.getLogChannel().getLogChannelId(), false);
                                previewLogMap.put(combi.stepMeta, logBuffer);
                            }
                        }
                    }
                }
            });
        }
    }

    private void checkErrorVisuals() {
        if (trans.getErrors() > 0) {
            StepStatusCodecMap.clear();
            for (StepMetaDataCombi combi : trans.getSteps()) {
                if (combi.step.getErrors() > 0) {
                    String channelId = combi.step.getLogChannel().getLogChannelId();
                    List<KettleLoggingEvent> eventList = KettleLogStore.getLogBufferFromTo(channelId, false, 0, KettleLogStore.getLastBufferLineNr());
                    StringBuilder logText = new StringBuilder();
                    for (KettleLoggingEvent event : eventList) {
                        Object message = event.getMessage();
                        if (message instanceof LogMessage) {
                            LogMessage logMessage = (LogMessage) message;
                            if (logMessage.isError()) {
                                logText.append(logMessage.getMessage()).append(Const.CR);
                            }
                        }
                    }
                    StepStatusCodecMap.put(combi.stepMeta, logText.toString());
                }
            }
        } else {
            StepStatusCodecMap.clear();
        }
    }


    public String[] getPerformanceSnapShotSteps() {
        if (executionConfiguration.isExecutingLocally()) {
            while (stepPerformanceSnapShotMap == null || stepPerformanceSnapShotMap.isEmpty()) {
                stepPerformanceSnapShotMap = trans.getStepPerformanceSnapShots();
                try {
                    Thread.sleep(100L);
                } catch (InterruptedException e) {
                    // Ignore errors
                }
            }
            Set<String> stepsSet = stepPerformanceSnapShotMap.keySet();
            String[] steps = stepsSet.toArray(new String[0]);
            Arrays.sort(steps);
            return steps;
        }
        return new String[]{};
    }

    public String[] runSteps() {
        while (previewDataMap == null || previewDataMap.isEmpty()) {
            try {
                Thread.sleep(100L);
            } catch (InterruptedException e) {
                // Ignore errors
            }
        }
        Set<StepMeta> stepMetaSet = previewDataMap.keySet();
        final StepMeta[] stepMetas = stepMetaSet.toArray(new StepMeta[0]);
        String[] stepNames = new String[stepMetas.length];
        for (int i = 0; i < stepMetas.length; i++)
            stepNames[i] = stepMetas[i].getName();
        Arrays.sort(stepNames);
        return stepNames;
    }

    public List<List<Long>> getStepPerformanceSnapShots(String stepName) {
        List<List<Long>> shots = new ArrayList<>();
        if (stepPerformanceSnapShotMap != null) {
            List<StepPerformanceSnapShot> stepPerformanceSnapShots = stepPerformanceSnapShotMap.get(stepName);
            if (stepPerformanceSnapShots != null && stepPerformanceSnapShots.size() > 1) {
                for (StepPerformanceSnapShot stepPerformanceSnapShot : stepPerformanceSnapShots) {
                    List<Long> shot = new ArrayList<>();
                    shot.add(stepPerformanceSnapShot.getLinesRead());
                    shot.add(stepPerformanceSnapShot.getLinesWritten());
                    shot.add(stepPerformanceSnapShot.getLinesInput());
                    shot.add(stepPerformanceSnapShot.getLinesOutput());
                    shot.add(stepPerformanceSnapShot.getLinesUpdated());
                    shot.add(stepPerformanceSnapShot.getLinesRejected());
                    shot.add(stepPerformanceSnapShot.getInputBufferSize());
                    shot.add(stepPerformanceSnapShot.getOutputBufferSize());
                    shots.add(shot);
                }
            }
        }
        return shots;
    }


    public List<StepStatusCodec> getStepMeasure() throws Exception {
        List<StepStatusCodec> list = new LinkedList<>();
        if (executionConfiguration.isExecutingLocally()) {
            for (int i = 0; i < trans.nrSteps(); i++) {
                StepInterface baseStep = trans.getRunThread(i);
                StepStatus stepStatus = new StepStatus(baseStep);
                StepStatusCodec StepStatusCodec = addStepStatusCodec(stepStatus);
                list.add(StepStatusCodec);
            }
        } else if (executionConfiguration.isExecutingRemotely()) {
            SlaveServer remoteSlaveServer = executionConfiguration.getRemoteServer();
            SlaveServerTransStatus transStatus = remoteSlaveServer.getTransStatus(transMeta.getName(), carteObjectId, 0);
            List<StepStatus> stepStatusList = transStatus.getStepStatusList();
            for (StepStatus stepStatus : stepStatusList) {
                StepStatusCodec StepStatusCodec = addStepStatusCodec(stepStatus);
                list.add(StepStatusCodec);
            }
        } else if (executionConfiguration.isExecutingClustered()) {
            SlaveServer masterServer = transSplitter.getMasterServer();
            SlaveServer[] slaves = transSplitter.getSlaveTargets();
            Map<TransMeta, String> carteMap = transSplitter.getCarteObjectMap();
            SlaveServerTransStatus masterServerTransStatus = masterServer.getTransStatus(transMeta.getName(), carteMap.get(transSplitter.getMaster()), 0);
            List<StepStatus> masterServerTransStatusStepStatusList = masterServerTransStatus.getStepStatusList();
            for (StepStatus masterServerTransStatusStepStatus : masterServerTransStatusStepStatusList) {
                StepStatusCodec StepStatusCodec = addStepStatusCodec(masterServerTransStatusStepStatus);
                list.add(StepStatusCodec);
            }
            for (SlaveServer slaveServer : slaves) {
                SlaveServerTransStatus slaveServerTransStatus = slaveServer.getTransStatus(transMeta.getName(), carteMap.get(transSplitter.getSlaveTransMap().get(slaveServer)), 0);
                List<StepStatus> slaveServerTransStatusStepStatusList = slaveServerTransStatus.getStepStatusList();
                for (StepStatus slaveServerTransStatusStepStatus : slaveServerTransStatusStepStatusList) {
                    StepStatusCodec StepStatusCodec = addStepStatusCodec(slaveServerTransStatusStepStatus);
                    list.add(StepStatusCodec);
                }
            }
        }
        return list;
    }

    public String[] getExecutionLog() throws Exception {
        if (executionConfiguration.isExecutingLocally()) {
            StringBuilder sb = new StringBuilder();
            KettleLogLayout logLayout = new KettleLogLayout(true);
            List<String> childIds = LoggingRegistry.getInstance().getLogChannelChildren(trans.getLogChannelId());
            List<KettleLoggingEvent> logLines = KettleLogStore.getLogBufferFromTo(childIds, true, -1, KettleLogStore.getLastBufferLineNr());
            for (KettleLoggingEvent kettleLoggingEvent : logLines) {
                String line = logLayout.format(kettleLoggingEvent).trim();
                sb.append(line).append("\n");
            }
            return sb.toString().split("\r?\n");
        } else if (executionConfiguration.isExecutingRemotely()) {
            SlaveServer remoteSlaveServer = executionConfiguration.getRemoteServer();
            SlaveServerTransStatus transStatus = remoteSlaveServer.getTransStatus(transMeta.getName(), carteObjectId, 0);
            return transStatus.getLoggingString().split("\r?\n");
        } else if (executionConfiguration.isExecutingClustered()) {
            SlaveServer masterServer = transSplitter.getMasterServer();
            SlaveServer[] slaves = transSplitter.getSlaveTargets();
            Map<TransMeta, String> carteMap = transSplitter.getCarteObjectMap();
            SlaveServerTransStatus transStatus = masterServer.getTransStatus(transMeta.getName(), carteMap.get(transSplitter.getMaster()), 0);
            StringBuilder log = new StringBuilder(transStatus.getLoggingString());
            for (SlaveServer slaveServer : slaves) {
                SlaveServerTransStatus slaveServerTransStatus = slaveServer.getTransStatus(transMeta.getName(), carteMap.get(transSplitter.getSlaveTransMap().get(slaveServer)), 0);
                if (StringUtils.isNotEmpty(slaveServerTransStatus.getLoggingString())) {
                    log.append(transStatus.getLoggingString());
                }
            }
            return log.toString().split("\r?\n");
        }
        return new String[]{};
    }

    private StepStatusCodec addStepStatusCodec(StepStatus stepStatus) {
        StepStatusCodec StepStatusCodec = new StepStatusCodec();
        String[] transLogFields = stepStatus.getTransLogFields();
        StepStatusCodec.setStepname(transLogFields[1]);
        StepStatusCodec.setStepCopy(Integer.valueOf(transLogFields[2]));
        StepStatusCodec.setLinesRead(Long.valueOf(transLogFields[3]));
        StepStatusCodec.setLinesWritten(Long.valueOf(transLogFields[4]));
        StepStatusCodec.setLinesInput(Long.valueOf(transLogFields[5]));
        StepStatusCodec.setLinesOutput(Long.valueOf(transLogFields[6]));
        StepStatusCodec.setLinesUpdated(Long.valueOf(transLogFields[7]));
        StepStatusCodec.setLinesRejected(Long.valueOf(transLogFields[8]));
        StepStatusCodec.setErrors(Long.valueOf(transLogFields[9]));
        StepStatusCodec.setActivation(transLogFields[10]);
        StepStatusCodec.setTime(transLogFields[11]);
        StepStatusCodec.setSpeed(transLogFields[12]);
        StepStatusCodec.setType(transLogFields[13]);
        return StepStatusCodec;
    }

    public void stop() {
        if (executionConfiguration.isExecutingLocally()) {
            if (trans != null)
                trans.stopAll();
        } else if (executionConfiguration.isExecutingRemotely()) {
            SlaveServer remoteSlaveServer = executionConfiguration.getRemoteServer();
            try {
                remoteSlaveServer.stopTransformation(transMeta.getName(), carteObjectId);
            } catch (Exception e) {
                throw new EtlException(e.getMessage());
            }
        } else if (executionConfiguration.isExecutingClustered()) {

        }

    }

    public void pause() {
        if (trans != null)
            trans.pauseRunning();
    }

    public Map<StepMeta, String> getStepStatusCodecMap() {
        return StepStatusCodecMap;
    }


    public Map<StepMeta, RowMetaInterface> getPreviewMetaMap() {
        return previewMetaMap;
    }


    public Map<StepMeta, List<Object[]>> getPreviewDataMap() {
        return previewDataMap;
    }


    public Map<StepMeta, StringBuffer> getPreviewLogMap() {
        return previewLogMap;
    }


    public String getExecutionId() {
        return executionId;
    }


    public TransMeta getTransMeta() {
        return transMeta;
    }


    public Trans getTrans() {
        return trans;
    }


    public TransSplitter getTransSplitter() {
        return transSplitter;
    }


    public boolean isFinished() {
        return finished;
    }

    public long getErrCount() {
        return errCount;
    }

    public TransExecutionConfiguration getExecutionConfiguration() {
        return executionConfiguration;
    }

}
